package ar.uba.fi.tonyvaliente.tree;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.logging.Logger;

import ar.uba.fi.tonyvaliente.files.Persistor;
import ar.uba.fi.tonyvaliente.files.PersistorException;
import ar.uba.fi.tonyvaliente.files.RecordPosition;

/**
 * Persistor para objetos BPlusTreeNode.
 * Debe estar asociado a un BPlusTreeFileWrapper para poder deserializar
 * las referencias a otros nodos.
 * Realiza la serializacion/deserializacion de claves y valores a traves de
 * otros objetos Persistor que recibe al ser instanciado.
 * 
 * @author Diego Morello
 *
 * @param <K> Clase a la cual pertenecen las claves del nodo
 * @param <V> Clase a la cual pertenecen los valores del nodo
 */
class NodePersistor<K extends Comparable<K>, V> extends Persistor<BPlusTreeNode<K,V>> {
	
	private static Logger log = Logger.getLogger(NodePersistor.class.getName());
	
	private Persistor<K> keyPersistor;
	private Persistor<V> valuePersistor;
	private BPlusTreeFileWrapper<K,V> associatedWrapper;
	
	/**
	 * Crea una nueva instancia de la clase, con un Persistor para claves y uno
	 * para valores, ademas de un file wrapper al cual se asocia para poder
	 * deserializar referencias a nodos.
	 * 
	 * @author Diego Morello
	 * 
	 * @param keyPersistor Persistor para claves
	 * @param valuePersistor Persistor para valores
	 * @param wrapper File wrapper asociado
	 * @throws NullPointerException Si alguno de los parametros es null
	 */
	public NodePersistor(Persistor<K> keyPersistor, Persistor<V> valuePersistor,
			BPlusTreeFileWrapper<K,V> wrapper) {
		
		if (keyPersistor == null || valuePersistor == null || wrapper == null) {
			throw new NullPointerException("Error al inicializar NodePersistor");
		}
		
		this.keyPersistor = keyPersistor;
		this.valuePersistor = valuePersistor;
		this.associatedWrapper = wrapper;
	}
	
	/**
	 * Serializa un objeto BPlusTreeNode, que puede ser hoja o interno.
	 * La serializacion de claves y valores se hace a traves de los Persistors
	 * con los cuales fue instanciado. 
	 * 
	 * @author Diego Morello
	 * 
	 * @param node Nodo a serializar
	 * @return byte array con el nodo serializado
	 * @throws PersistorException si ocurre un error al serializar
	 */
	public byte[] serialize(BPlusTreeNode<K,V> node) throws PersistorException {
		
		log.fine("serialize IN");
		
		try{
			log.finer("node: " + node);
		}catch(Exception e){
			
		}
		
		if (node instanceof InnerBPlusTreeNode) {
			InnerBPlusTreeNode<K,V> innerNode = (InnerBPlusTreeNode<K,V>)node;
			return this.serialize(innerNode);
		} else {
			LeafBPlusTreeNode<K,V> leafNode = (LeafBPlusTreeNode<K,V>)node;
			return this.serialize(leafNode);
		}
	}
	
	/**
	 * Crea un BPlusTreeNode a partir del byte array recibido por parametro.
	 * La deserializacion de claves y valores se hace a traves de los Persistors
	 * con los cuales fue instanciado.
	 * 
	 * @author Diego Morello
	 * 
	 * @param bytes byte array con el nodo serializado
	 * @return Objeto de clase InnerBPlusTreeNode o LeafBPlusTreeNode
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	public BPlusTreeNode<K,V> deserialize(byte[] bytes) throws PersistorException {
		boolean isInnerNode = (bytes[0] != 0);
		if (isInnerNode) {
			return this.deserializeInnerNode(bytes);
		} else {
			return this.deserializeLeafNode(bytes);
		}
	}
	
	/**
	 * Serializa un nodo interno.
	 * El array de bytes resultante tiene la siguiente estructura:
	 * 
	 * - flag indicador de nodo interno ....... 1 byte (distinto de cero)
	 * - cantidad de elementos ................ 4 bytes
	 * - maxima cantidad de elementos ......... 4 bytes
	 * - numero de bloque del nodo padre ...... 4 bytes (< 0 para nodo raiz)
	 * - offset del nodo padre en el bloque ... 4 bytes (< 0 para nodo raiz)
	 * - cantidad de claves ................... 4 bytes
	 * Por cada clave:
	 * - cantidad de bytes de la clave ........ 4 bytes
	 * - bytes de la clave .................... *
	 * Por cada referencia a nodo hijo:
	 * - numero de bloque del nodo hijo ....... 4 bytes
	 * - offset del nodo hijo en el bloque .... 4 bytes
	 * 
	 * @param node
	 * @return array de bytes con el nodo interno serializado
	 * @throws PersistorException si ocurre un error al serializar
	 */
	private byte[] serialize(InnerBPlusTreeNode<K,V> node) throws PersistorException {
		
		log.fine("serialize innerNode IN");
		
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream out = new DataOutputStream(baos);
		try {
			// Obtengo "datos de cabecera" del nodo
			int cantElementos = node.getCantidadElementos();
			int maxCantElementos = node.getMaxCantidadElementos();
			
			// Escribo flag indicador de nodo interno (TRUE)
			out.writeByte((byte)1);
			
			// Escribo los datos de cabecera
			out.writeInt(cantElementos);
			out.writeInt(maxCantElementos);

			// Escribo los datos del nodo padre en el byte array
			DiskReference<K,V> parent = (DiskReference<K,V>)node.getParent();
			this.writeDiskReference(parent, out);
			
			// Escribo claves
			List<K> keys = node.getKeys();
			out.writeInt(keys.size());
			for (K key: keys) {
				this.writeKey(key, out);
			}
			
			// Escribo las referencias a nodos hijos
			List<Reference<K,V>> references = node.getReferences();
			// TODO: quitar esto luego del debug
			assert(keys.size() + 1 == references.size());
			for (Reference<K,V> ref: references) {
				DiskReference<K,V> diskRef = (DiskReference<K,V>)ref;
				this.writeDiskReference(diskRef, out);
			}
			
			log.fine("nodo serializado");
			
			out.close();
		} catch (IOException e) {
			throw new PersistorException("Error al serializar nodo interno", e);
		} catch (ClassCastException e) {
			throw new PersistorException("Error al serializar nodo interno: "
					+ "alguna referencia no apunta a un nodo en disco", e);
		} catch (AssertionError e) {
			// TODO: quitar esto luego del debug
			throw new PersistorException("Error al serializar nodo interno: "
					+ "la cantidad de referencias a nodos es incorrecta", e);
		}
		
		return baos.toByteArray();
	}
	
	/**
	 * Crea un objeto InnerBPlusTreeNode a partir de un byte array.
	 * Espera la siguiente estructura en dicho byte array:
	 * 
	 * - flag indicador de nodo interno ....... 1 byte (distinto de cero)
	 * - cantidad de elementos ................ 4 bytes
	 * - maxima cantidad de elementos ......... 4 bytes
	 * - numero de bloque del nodo padre ...... 4 bytes (< 0 para nodo raiz)
	 * - offset del nodo padre en el bloque ... 4 bytes (< 0 para nodo raiz)
	 * - cantidad de claves ................... 4 bytes
	 * Por cada clave:
	 * - cantidad de bytes de la clave ........ 4 bytes
	 * - bytes de la clave .................... *
	 * Por cada referencia a nodo hijo:
	 * - numero de bloque del nodo hijo ....... 4 bytes
	 * - offset del nodo hijo en el bloque .... 4 bytes
	 * 
	 * @param bytes array de bytes a partir del cual crear el nodo interno
	 * @return Objeto InnerBPlusTreeNode deserializado
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private InnerBPlusTreeNode<K,V> deserializeInnerNode(byte[] bytes)
			throws PersistorException {
		
		log.fine("deserializeInnerNode IN");
		
		InnerBPlusTreeNode<K,V> node = null;
		int cantElementos;
		int maxCantElementos;
		DiskReference<K,V> parentRef;
		ArrayList<K> keys;
		ArrayList<Reference<K,V>> references;
		
		ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
		DataInputStream in = new DataInputStream(bais);
		try {
			// Verifico que los bytes correspondan a un nodo interno
			byte innerNodeFlag = in.readByte();
			if (innerNodeFlag == 0) {
				throw new PersistorException("Error al deserializar nodo interno:"
						+ "el byte array leido corresponde a un nodo hoja");
			}
			
			// Leo cantidad de elementos
			cantElementos = in.readInt();
			if (cantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo interno: "
						+ "la cantidad de elementos no puede ser negativa");
			}
			
			// Leo cantidad maxima de elementos
			maxCantElementos = in.readInt();
			if (maxCantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo interno: "
						+ "la cantidad maxima de elementos no puede ser negativa");
			} else if (maxCantElementos < cantElementos) {
				throw new PersistorException("Error al deserializar nodo interno: "
						+ "la cantidad maxima de elementos no puede ser menor "
						+ "a la cantidad real de elementos");
			}
			
			// Leo la referencia al nodo padre
			parentRef = this.readDiskReference(in);
			
			// Leo la cantidad de claves
			int cantidadClaves = in.readInt();
			if (cantidadClaves < 0) {
				throw new PersistorException("Error al deserializar nodo interno: "
						+ "la cantidad de claves en el nodo no puede ser negativa");
			}
			
			// Leo las claves
			keys = new ArrayList<K>();
			for (int i = 0; i < cantidadClaves; i++) {
				K key = this.readKey(in);
				keys.add(key);
			}
			
			// Leo las referencias
			int cantidadRef = (cantidadClaves > 0)? cantidadClaves + 1 : 0;
			references = new ArrayList<Reference<K,V>>();
			for (int i = 0; i < cantidadRef; i++) {
				DiskReference<K,V> ref = this.readDiskReference(in);
				references.add(ref);
			}
			
			// Armo el nodo
			node = new InnerBPlusTreeNode<K,V>(parentRef);
			node.setMaxCantidadElementos(maxCantElementos);
			node.setCantidadElementos(cantElementos);
			node.setKeys(keys);
			node.setReferences(references);
			
			log.finer("deserializeInnerNode OUT");
			try{
				log.finer("innerNode: " + node);
			}catch(Exception e){
				
			}
			
			in.close();
			
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar nodo interno", e);
		}
		
		return node;
	}
	
	/**
	 * Serializa un nodo hoja.
	 * El byte array resultante tiene la siguiente estructura:
	 * 
	 * - flag indicador de nodo interno .......... 1 byte (cero)
	 * - cantidad de elementos ................... 4 bytes
	 * - maxima cantidad de elementos ............ 4 bytes
	 * - numero de bloque del nodo padre ......... 4 bytes (< 0 para nodo raiz)
	 * - offset del nodo padre en el bloque ...... 4 bytes (< 0 para nodo raiz)
	 * - cantidad de pares clave-valor ........... 4 bytes
	 * - numero de bloque del nodo previo ........ 4 bytes
	 * - offset del nodo previo en el bloque ..... 4 bytes
	 * - numero de bloque del nodo siguiente ..... 4 bytes
	 * - offset del nodo siguiente en el bloque .. 4 bytes
	 * Por cada par clave-valor
	 * - cantidad de bytes de la clave ........... 4 bytes
	 * - bytes de la clave ....................... *
	 * - cantidad de bytes deL valor ............. 4 bytes
	 * - bytes del valor ......................... *
	 * 
	 * @param node
	 * @return array de bytes con el nodo interno serializado
	 * @throws PersistorException si ocurre un error al serializar
	 */
	private byte[] serialize(LeafBPlusTreeNode<K,V> node) throws PersistorException {
		
		log.fine("serialize leafNode IN");
		
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream out = new DataOutputStream(baos);
		try {
			// Obtengo "datos de cabecera" del nodo
			int cantidadElementos = node.getCantidadElementos();
			int maxCantidadElementos = node.getMaxCantidadElementos();
			
			// Escribo flag indicador de nodo hoja (FALSE)
			out.writeByte((byte)0);
			
			// Escribo los datos de cabecera
			out.writeInt(cantidadElementos);
			out.writeInt(maxCantidadElementos);
			
			// Escribo los datos del nodo padre
			DiskReference<K,V> parentRef = (DiskReference<K,V>)node.getParent();
			this.writeDiskReference(parentRef, out);
			
			// Obtengo y escribo datos de nodo previo
			DiskReference<K,V> previous = (DiskReference<K,V>)node.getPrevious();
			this.writeDiskReference(previous, out);
			
			// Obtengo y escribo datos de nodo siguiente
			DiskReference<K,V> next = (DiskReference<K,V>)node.getNext();
			this.writeDiskReference(next, out);
			
			// Obtengo y escribo la cantidad de pares clave-valor
			Map<K,V> datos = node.getDatos();
			out.writeInt(datos.size());
			
			// Obtengo y escribo claves y valores
			Iterator<K> it = datos.keySet().iterator();
			while (it.hasNext()) {
				K key = it.next();
				this.writeKey(key, out);
				
				V value = datos.get(key);
				this.writeValue(value, out);
			}
		
			log.fine("nodo serializado");
			
			out.close();
		} catch (IOException e) {
			throw new PersistorException("Error al serializar nodo hoja");
		} catch (ClassCastException e) {
			throw new PersistorException("Error al serializar nodo hoja: "
					+ "alguna referencia no apunta a un nodo en disco", e);
		}
		
		return baos.toByteArray();
	}
	
	/**
	 * Crea un objeto LeafBPlusTreeNode a partir de un byte array.
	 * Espera la siguiente estructura en dicho byte array:
	 * 
	 * - flag indicador de nodo interno .......... 1 byte (cero)
	 * - cantidad de elementos ................... 4 bytes
	 * - maxima cantidad de elementos ............ 4 bytes
	 * - numero de bloque del nodo padre ......... 4 bytes (< 0 para nodo raiz)
	 * - offset del nodo padre en el bloque ...... 4 bytes (< 0 para nodo raiz)
	 * - numero de bloque del nodo previo ........ 4 bytes
	 * - offset del nodo previo en el bloque ..... 4 bytes
	 * - numero de bloque del nodo siguiente ..... 4 bytes
	 * - offset del nodo siguiente en el bloque .. 4 bytes
	 * - cantidad de pares clave-valor ........... 4 bytes
	 * Por cada par clave-valor
	 * - cantidad de bytes de la clave ........... 4 bytes
	 * - bytes de la clave ....................... *
	 * - cantidad de bytes deL valor ............. 4 bytes
	 * - bytes del valor ......................... *
	 * 
	 * @param bytes array de bytes a partir del cual crear el nodo hoja
	 * @return Objeto LeafBPlusTreeNode deserializado
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private LeafBPlusTreeNode<K,V> deserializeLeafNode(byte[] bytes) throws PersistorException {
		
		log.fine("deserializeLeafNode IN");
		
		LeafBPlusTreeNode<K,V> node = null;
		int cantElementos;
		int maxCantElementos;
		DiskReference<K,V> parentRef;
		DiskReference<K,V> previous;
		DiskReference<K,V> next;
		TreeMap<K,V> datos;
		
		ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
		DataInputStream in = new DataInputStream(bais);
		try {
			// Verifico que el byte array recibido corresponda a un nodo hoja
			byte innerNodeFlag = in.readByte();
			if (innerNodeFlag != 0) {
				throw new PersistorException("Error al deserializar nodo hoja: "
						+ "el byte array leido corresponde a un nodo interno");
			}
			
			// Leo cantidad de elementos
			cantElementos = in.readInt();
			if (cantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo hoja: "
						+ "la cantidad de elementos no puede ser negativa");
			}
			
			// Leo cantidad maxima de elementos
			maxCantElementos = in.readInt();
			if (maxCantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo hoja: "
						+ "la cantidad maxima de elementos no puede ser negativa");
			} else if (maxCantElementos < cantElementos) {
				throw new PersistorException("Error al deserializar nodo hoja: "
						+ "la cantidad maxima de elementos no puede ser menor "
						+ "a la cantidad real de elementos");
			}
			
			// Leo la referencia al nodo padre
			parentRef = this.readDiskReference(in);
			
			// Leo la referencia al nodo previo
			previous = this.readDiskReference(in);

			// Leo la referencia al nodo siguiente
			next = this.readDiskReference(in);
			
			// Obtengo la cantidad de claves y valores
			int size = in.readInt();
			if (size < 0) {
				throw new PersistorException("Error al deserializar nodo hoja: "
						+ "la cantidad de pares clave-valor no puede ser negativa");
			}
			
			// Leo claves y valores
			datos = new TreeMap<K,V>();
			for (int i = 0; i < size; i++) {
				K key = this.readKey(in);
				V value = this.readValue(in);
				datos.put(key, value);
			}
			
			in.close();
			
			// Armo el nodo
			node = new LeafBPlusTreeNode<K,V>(parentRef, maxCantElementos);
			node.setPrevious(previous);
			node.setNext(next);
			node.setDatos(datos);
			node.setCantidadElementos(cantElementos);
			
			log.fine("deserializeLeafNode OUT");
			try{
				log.finer("nodo obtenido: " + node);
			}catch(Exception e){
				
			}
			
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar nodo hoja", e);
		}
		
		
		return node;
	}
	
	/**
	 * Escribe una DiskReference en el DataOutputStream recibido
	 * @param ref referencia a escribir
	 * @param out DataOutputStream donde escribir la referencia
	 * @throws PersistorException si ocurre algun error al escribir
	 */
	private void writeDiskReference(DiskReference<K,V> ref, DataOutputStream out) throws PersistorException {
		try {
			if (ref != null) {
				out.writeInt(ref.getRecordPosition().getBlockNumber());
				out.writeInt(ref.getRecordPosition().getOffset());
			} else {
				out.writeInt(-1);
				out.writeInt(-1);
			}
		} catch (IOException e) {
			throw new PersistorException("Error al serializar referencia", e);
		}
	}
	
	/**
	 * Lee una DiskReference del DataInputStream recibido. 
	 * @param in
	 * @return objeto DiskReference
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private DiskReference<K,V> readDiskReference(DataInputStream in)
			throws PersistorException {
		
		try {
			int blockNumber = in.readInt();
			int offset = in.readInt();
		
			if (blockNumber >= 0 && offset >= 0) {
				DiskReference<K,V> ref = new DiskReference<K,V>(associatedWrapper);
				ref.setRecordPosition(new RecordPosition(blockNumber, offset));
				return ref;
			} else if (blockNumber < 0 && offset < 0) {
				return null;
			} else {
				throw new PersistorException("Error al deserializar referencia a nodo: "
						+ "la posicion leida no es valida");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar referencia a nodo", e);
		}
	}
	
	/**
	 * Escribe una clave en el DataOutputStream recibido por parametro
	 * @param key clave a escribir
	 * @param out DataOutputStream donde escribir
	 * @throws PersistorException si ocurre un error al escribir
	 */
	private void writeKey(K key, DataOutputStream out) throws PersistorException {
		try {
			byte[] keyBytes = keyPersistor.serialize(key);
			out.writeInt(keyBytes.length);
			out.write(keyBytes);
		} catch (IOException e) {
			throw new PersistorException("Error al escribir clave");
		}
	}
	
	/**
	 * Deserializa una clave a partir del DataInputStream recibido
	 * @param in
	 * @return clave (objeto de clase K)
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private K readKey(DataInputStream in) throws PersistorException {
		try {
			int keySize = in.readInt();
			if (keySize >= 0) {
				byte[] keyBytes = new byte[keySize];
				if (in.read(keyBytes) != keySize) {
					throw new PersistorException("Error al deserializar clave: "
						+ "numero erroneo de bytes leidos");
				}
				return keyPersistor.deserialize(keyBytes);
			} else {
				throw new PersistorException("Error al deserializar clave: "
						+ "el tamanio en bytes no puede ser negativo");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar clave");
		}
	}
	
	/**
	 * Escribe un valor en el DataOutputStream recibido por parametro
	 * @param value valor a escribir
	 * @param out DataOutputStream donde escribir
	 * @throws PersistorException si ocurre un error al escribir
	 */
	private void writeValue(V value, DataOutputStream out) throws PersistorException {
		try {
			byte[] valueBytes = valuePersistor.serialize(value);
			out.writeInt(valueBytes.length);
			out.write(valueBytes);
		} catch (IOException e) {
			throw new PersistorException("Error al escribir valor");
		}
	}
	
	/**
	 * Deserializa un valor a partir del DataInputStream recibido
	 * @param in
	 * @return valor (objeto de clase V)
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private V readValue(DataInputStream in) throws PersistorException {
		try {
			int valueSize = in.readInt();
			if (valueSize >= 0) {
				byte[] valueBytes = new byte[valueSize];
				if (in.read(valueBytes) != valueSize) {
					throw new PersistorException("Error al deserializar valor: "
						+ "numero erroneo de bytes leidos");
				}
				return valuePersistor.deserialize(valueBytes);
			} else {
				throw new PersistorException("Error al deserializar valor: "
						+ "el tamanio en bytes no puede ser negativo");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar valor");
		}
	}
}
